package com.caine.etcd.tool;

import com.caine.etcd.hanlder.WatchHandler;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.options.LeaseOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.options.WatchOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchEvent.EventType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Component
public class EtcdTools {

    @Resource
    private Watch watch;
    @Resource
    private Lease lease;
    @Resource
    private KV kv;

    /**
     * key 监听
     *
     * @param key
     * @param handler
     */
    public void addWatch(String key, WatchHandler handler) {
        WatchOption.Builder builder = WatchOption.newBuilder().withProgressNotify(true).withPrevKV(true).withNoPut(true);
        WatchOption watchOption = builder.build();
        watch.watch(ByteSequence.from(key, Charset.forName("UTF-8")), watchOption, c -> {
            List<WatchEvent> events = c.getEvents();
            events.forEach(e -> {
                KeyValue keyValue = e.getPrevKV();
                String k = bs2Str(keyValue.getKey());
                String v = bs2Str(keyValue.getValue());
                if (e.getEventType() == EventType.DELETE) {
                    long leaseId = e.getKeyValue().getLease();
                    long ttl = 0;
                    try {
                        ttl = lease.timeToLive(leaseId, LeaseOption.DEFAULT).get().getTTl();
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                    } catch (ExecutionException executionException) {
                        executionException.printStackTrace();
                    }
                    if (handler != null) {
                        if (ttl > 0) {
                            handler.delete(k, v);
                        } else {
                            handler.expire(k, v);
                        }
                    }
                } else if (e.getEventType() == EventType.PUT) {
                    if (handler != null) {
                        handler.put(k, v);
                    }
                }
            });
        });
    }

    /**
     * 设置值
     *
     * @param key
     * @param val
     */
    public void put(String key, String val) {
        kv.put(str2Bs(key), str2Bs(val));
    }

    /**
     * 设置值,带过期时间
     *
     * @param key
     * @param val
     * @param expire
     */
    public void put(String key, String val, long expire) {
        try {
            long leaseId = lease.grant(expire).get().getID();
            PutOption putOption = PutOption.newBuilder().withLeaseId(leaseId).build();
            kv.put(str2Bs(key), str2Bs(val), putOption);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

    private String bs2Str(ByteSequence bs) {
        return new String(bs.getBytes(), Charset.forName("UTF-8"));
    }

    private ByteSequence str2Bs(String str) {
        return ByteSequence.from(str, Charset.forName("UTF-8"));
    }
}
